package com.atguigu.cdc;


import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class Flink_CDC {
    public static void main(String[] args) throws Exception {

        // create
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

//        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall2022/"));
//        env.enableCheckpointing(5000L);


//        MySqlSource<String> mysqlSource = MySqlSource.<String>builder()
//                .hostname("hadoop102")
//                .port(3306)
//                .username("root")
//                .password("1234")
//                .databaseList("gmall2022")
//                .tableList("gmall2022.base_trademark")
//                .deserializer(new StringDebeziumDeserializationSchema())
//                .startupOptions(StartupOptions.initial())
//                .build();

        DebeziumSourceFunction<String> build = MySqlSource
                .<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("1234")
                .databaseList("gmall2022")
                .tableList("gmall2022.base_trademark")
                .deserializer(new StringDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

//        DataStreamSource<String> stream = env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "MysqlSource");

        DataStreamSource<String> stream = env.addSource(build);
        stream.print();

        env.execute();

    }
}
